package com.zb.emq;

import com.alibaba.fastjson.JSON;
import com.zb.dto.DeviceInfoDTO;
import com.zb.dto.DeviceLocation;
import com.zb.es.EsRepository;
import com.zb.service.*;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Map;

@Component
@Slf4j
public class EmqMsgProcess implements MqttCallback {

    @Autowired
    private TbQuotaService quotaService;
    @Autowired
    private TbAlarmService alarmService;
    @Autowired
    private DeviceService deviceService;
    @Autowired
    private TbGpsService tbGpsService;
    @Autowired
    private EsRepository esRepository;
    @Autowired
    private NoticeService noticeService;

    @Override
    public void connectionLost(Throwable throwable) {
        log.info("重新建立连接...");
    }

    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
        log.info("接收mqtt服务器发送的主题：" + topic);
        String payload = new String(mqttMessage.getPayload());
        System.out.println("接收的数据是:" + payload);
        Map map = JSON.parseObject(payload, Map.class);
        //区分不同的主题
        if (topic.equals("geo")) {
            //解析GPS
            DeviceLocation deviceLocation = tbGpsService.analysis(topic, map);
            System.out.println(deviceLocation);
            //保存设备到es
            esRepository.saveLocation(deviceLocation);
            //透传GPS数据
            noticeService.gpsTransfer(deviceLocation);
        } else {
            //解析指标
            DeviceInfoDTO infoDTO = quotaService.analysis(topic, map);
            //验证报警
            DeviceInfoDTO deviceInfoDTO = alarmService.verifyDeviceInfo(infoDTO);
            System.out.println(JSON.toJSONString(deviceInfoDTO));
            //保存设备
            deviceService.saveDeviceInfo(deviceInfoDTO.getDeviceDTO());
            //添加数据到influxDB
            quotaService.saveQuotaToInflux(deviceInfoDTO.getQuotaList());
            //透传数据
            noticeService.quotaTransfer(deviceInfoDTO.getQuotaList());
        }
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        log.info("断开连接");
    }
}
